[ISSUE #10988] Fix concurrent commitOffset race losing queue offsets for a new topic@group - #10992
[ISSUE #10988] Fix concurrent commitOffset race losing queue offsets for a new topic@group#10992unbridled-41 wants to merge 1 commit into
Conversation
…fsets for a new topic@group
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR fixes a race condition in ConsumerOffsetManager.commitOffset where concurrent commits for different queues of a new topic@group could overwrite each other's entries. The original check-then-act pattern (if null, create and put) is replaced with the atomic computeIfAbsent, ensuring only one map is created per key. A comprehensive test simulates concurrent commits from 8 threads across 500 keys to verify the fix.
LGTM — excellent fix for the concurrency bug!
Automated review by github-manager-bot
Test evidence (before → after)
On the unfixed code: 5/5 runs fail. Example failure (queues 0 and 7 of The race window is the very first With this PR: 5/5 runs pass ( The loss is not just a test artifact: every commit the losing thread made for that queue is gone until that queue's next commit (5s flush cadence, but Side note on CI: the workflow runs for this PR are in |
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
Fixes a race condition in commitOffset where concurrent commits for different queues of a new topic@group could overwrite each other's entries, causing queue offset data loss.
Observations
- ✅ Replaces manual get-put pattern with atomic
computeIfAbsent, eliminating the race window - ✅ The comment clearly explains the concurrency issue: concurrent commits for different queues of a new entry could overwrite each other
- ✅ Includes comprehensive test
testConcurrentCommitOffsetDoesNotLoseQueuesthat verifies 8 threads committing 500 keys concurrently don't lose queue entries - The fix is minimal, correct, and well-tested
Analysis
The original code had a classic check-then-act race:
map = this.offsetTable.get(key);
if (null == map) {
map = new ConcurrentHashMap<>(2);
map.put(queueId, offset);
this.offsetTable.put(key, map); // ← race: another thread may have already put a map
}The fix uses computeIfAbsent which is atomic:
ConcurrentMap<Integer, Long> map = this.offsetTable.computeIfAbsent(key,
k -> new ConcurrentHashMap<>(2));
Long storeOffset = map.put(queueId, offset);This ensures all concurrent queue entries are preserved. Excellent fix.
LGTM — important concurrency bug fix.
Automated review by github-manager-bot
2 participants
Which Issue(s) This PR Fixes
Brief Description
ConsumerOffsetManager#commitOffsetused a non-atomic check-then-act to create the pertopic@groupmap:When two remoting threads commit offsets for different queues of the same brand-new
topic@groupconcurrently (the normal situation right after a consumer group starts on a multi-queue topic, or while a pop consumer acks several queues), both seenull, both install their own map, and the secondputsilently drops the first queue's offset.queryOffsetthen returns-1for that queue until its next commit, so a consumer reconnect/restart in the window re-initializes perconsumeFromWhere(duplicate consumption or skipping to max). The same race re-arms afterremoveOffset(group)/cleanOffsetByTopicwhile the group is still consuming.The fix replaces the get/put pair with
offsetTable.computeIfAbsent(key, k -> new ConcurrentHashMap<>(2))— the same idiom the class already uses incommitPullOffset— so concurrent commits always land in the same map. The less-than-store warn behavior is unchanged.How Did You Test This Change?
Added
ConsumerOffsetManagerTest#testConcurrentCommitOffsetDoesNotLoseQueues: 8 threads commit a distinct queue for 500 brand-newtopic@groupkeys each, then the test asserts every key retains all 8 queue offsets. It fails reproducibly on the unfixed code and passes with this change.mvn -pl broker test -Dtest=ConsumerOffsetManagerTestpasses (6/6).